﻿import unittest, sys, datetime, sqlite3 as sqlite

sys.path.append('')
from PTable import *


class Test_PTableTest(unittest.TestCase):
    """
    Тестирование класса PTable.
    """

    @classmethod
    def setUpClass(cls):
        cls._connection = sqlite.connect('testdb')
        cls._tableName = 'test'
        cls._columnsNames = ('name', 'isgood')
        cursor = cls._connection.cursor()
        cursor.execute('CREATE TABLE IF NOT EXISTS {} (id INTEGER PRIMARY KEY, {} TEXT, {} INTEGER)'.format(
            cls._tableName, cls._columnsNames[0], cls._columnsNames[1]
        ))
        cursor.execute('DELETE FROM {}'.format(cls._tableName))  # Прежде всего, очищаем талблицу от всего.
        cls._connection.commit()

    @classmethod
    def tearDownClass(cls):
        cls._connection.rollback()
        cls._connection.close()

    def tearDown(self):
        # Очистка  тестируемой таблицы.
        cursor = self._connection.cursor()
        cursor.execute('DELETE FROM {}'.format(self._tableName))
        self._connection.commit()

    def test_addRow(self):
        cursor = Test_PTableTest._connection.cursor()
        testTable = PTable(Test_PTableTest._tableName, Test_PTableTest._connection)

        testTable.addRow([1, 'name1', 1])
        testTable.addRow((2, 'name2', 0))
        testTable.addRow({
            'id': 100,
            Test_PTableTest._columnsNames[0]: 'name3',
            Test_PTableTest._columnsNames[1]: 1
        })

        cursor.execute('SELECT {}, {} FROM {}'.format(
            Test_PTableTest._columnsNames[0], Test_PTableTest._columnsNames[1], Test_PTableTest._tableName
        ))
        results = cursor.fetchall()

        ethalons = [
            ('name1', 1),
            ('name2', 0),
            ('name3', 1)
        ]

        for pair in zip(ethalons, results):
            self.assertEqual(pair[0], pair[1], 'Failed with {}.'.format(pair))

    def test_getAllRows(self):
        testTable = PTable(Test_PTableTest._tableName, Test_PTableTest._connection)

        testTable.addRow((100, 'name01', 0))
        testTable.addRow((200, 'name02', 1))
        testTable.addRow((300, 'name03', 0))
        testTable.addRow((400, 'name04', 1))

        results = testTable.getAllRows()

        ethalons = [
            (100, 'name01', 0),
            (200, 'name02', 1),
            (300, 'name03', 0),
            (400, 'name04', 1)
        ]

        for pair in zip(ethalons, results):
            self.assertEqual(pair[0], pair[1], 'Failed with {}.'.format(pair))

    def test_deleteRow(self):
        testTable = PTable(Test_PTableTest._tableName, Test_PTableTest._connection)

        testTable.addRow((100, 'name01', 0))
        testTable.addRow((200, 'name02', 1))
        testTable.addRow((300, 'name03', 0))
        testTable.addRow((400, 'name04', 1))

        testTable.deleteRow(200)
        testTable.deleteRow(300)

        results = testTable.getAllRows()

        ethalons = [
            (100, 'name01', 0),
            (400, 'name04', 1)
        ]
        for pair in zip(ethalons, results):
            self.assertEqual(pair[0], pair[1], 'Failed with {}.'.format(pair))

    def test_changeRow(self):
        testTable = PTable(Test_PTableTest._tableName, Test_PTableTest._connection)

        testTable.addRow((100, 'name01', 0))
        testTable.changeRow(100, {'name': 'name0100', 'isgood': 1})

        ethalon = [(100, 'name0100', 1)]
        results = testTable.getAllRows()

        for pair in zip(ethalon, results):
            self.assertEqual(pair[0], pair[1], 'Failed with {}.'.format(pair))

    def test_select(self):
        cursor = Test_PTableTest._connection.cursor()
        cursor.execute(
            'CREATE TABLE IF NOT EXISTS test_find ('
            'id INTEGER PRIMARY KEY, a INTEGER, b REAL, c TEXT, d BLOB, e DATE, f TIMESTAMP)'
        )
        cursor.execute('DELETE FROM test_find')
        Test_PTableTest._connection.commit()

        testTable = PTable('test_find', Test_PTableTest._connection)
        testTable.addRow({
            'a': 15,
            'b': 0.7,
            'c': 'Test1',
            'd': b'bytes1',
            'e': datetime.date(2013, 4, 19),
            'f': datetime.datetime.now()
        })
        testTable.addRow({
            'a': 150,
            'b': 1.17,
            'c': 'Test2',
            'd': b'bytes2',
            'e': datetime.date(2013, 7, 19),
            'f': datetime.datetime(2014, 7, 19, 15, 29, 43, 79060)
        })
        testTable.addRow({
            'a': 1500,
            'b': 2.7,
            'c': 'Test3',
            'd': b'bytes3',
            'e': datetime.date(2013, 8, 19),
            'f': datetime.datetime.now()
        })
        testTable.addRow({
            'a': 0,
            'b': 12.7,
            'c': 'Test4',
            'd': b'bytes4',
            'e': datetime.date(2013, 9, 1),
            'f': datetime.datetime(2013, 9, 1, 8, 30, 0, 0)
        })
        testTable.addRow({
            'a': 10,
            'b': 10.0,
            'c': 'Test5',
            'd': b'bytes5',
            'e': datetime.date(2013, 9, 1),
            'f': datetime.datetime(2013, 9, 1, 8, 30, 0, 0)
        })

        ethalons = [
            [(15, 0.7, 'Test1'), (10,  10, 'Test5')],
            [(1500, 2.7, 'Test3', b'bytes3')],
            [('Test4',)],
            [(4,), (5,)],
            [('Test2',), ('Test3',)]
        ]
        conditions = [
            {'a': (10, 20)},
            {'b': (0, 3), 'd': b'bytes3'},
            {'c': (None, 'Test2'), 'a': (10, None)},
            {'e': datetime.date(2013, 9, 1)},
            {'e': (datetime.date(2013, 4, 29), datetime.date(2013, 8, 29))}
        ]
        selects = [
            ('a', 'b', 'c'),
            ('a', 'b', 'c', 'd'),
            ('c',),
            ('id',),
            ('c',)
        ]
        orders = [
            tuple(),
            ('a',),
            tuple(),
            ('id', 'a'),
            tuple()
        ]
        results = []
        for (c, s, o) in zip(conditions, selects, orders):
            results.append(testTable.select(c, s, o))

        for pair in zip(ethalons, results):
            self.assertEqual(pair[0], pair[1], 'Failed with {}.'.format(pair))

    def test_fetchColumnsWithoutKeys(self):
        cursor = Test_PTableTest._connection.cursor()
        cursor.executescript("""
            CREATE TABLE IF NOT EXISTS test_wo_keys_1 (id INTEGER PRIMARY KEY, a1 INTEGER, b1 REAL, c1 TEXT, d1 BLOB, e1 DATE, f1 TIMESTAMP);
            CREATE TABLE IF NOT EXISTS test_wo_keys_2 (a2 INTEGER, b2 INTEGER, c2 INTEGER, PRIMARY KEY(a2, b2), FOREIGN KEY(c2) REFERENCES `test_wo_keys_1`(id));
            CREATE TABLE IF NOT EXISTS test_wo_keys_3 (id INTEGER, b3 INTEGER, c3 INTEGER, d3 BOOL, PRIMARY KEY(id), FOREIGN KEY(b3) REFERENCES `test_wo_keys_1`(id), FOREIGN KEY(c3) REFERENCES `test_wo_keys_2`(a2));
            """
        )
        Test_PTableTest._connection.commit()

        ethalons = [
            [(1, 'a1', 'INTEGER'), (2, 'b1', 'REAL'), (3, 'c1', 'TEXT'), (4, 'd1', 'BLOB'), (5, 'e1', 'DATE'), (6, 'f1', 'TIMESTAMP')],
            [],
            [(3, 'd3', 'BOOL')]
        ]
        results = []
        for name in ['test_wo_keys_1', 'test_wo_keys_2', 'test_wo_keys_3']:
            testTable = PTable(name, Test_PTableTest._connection)
            results.append(testTable.fetchColumnsWithoutKeys())

        for pair in zip(ethalons, results):
            self.assertEqual(pair[0], pair[1], 'Failed with {}.'.format(pair))

    def test_fetchColumnsWithoutForeignKeys(self):
        cursor = Test_PTableTest._connection.cursor()
        cursor.executescript("""
            CREATE TABLE IF NOT EXISTS test_wo_keys_1 (id INTEGER PRIMARY KEY, a1 INTEGER, b1 REAL, c1 TEXT, d1 BLOB, e1 DATE, f1 TIMESTAMP);
            CREATE TABLE IF NOT EXISTS test_wo_keys_2 (a2 INTEGER, b2 INTEGER, c2 INTEGER, PRIMARY KEY(a2, b2), FOREIGN KEY(c2) REFERENCES `test_wo_keys_1`(id));
            CREATE TABLE IF NOT EXISTS test_wo_keys_3 (id INTEGER, b3 INTEGER, c3 INTEGER, d3 BOOL, PRIMARY KEY(id), FOREIGN KEY(b3) REFERENCES `test_wo_keys_1`(id), FOREIGN KEY(c3) REFERENCES `test_wo_keys_2`(a2));
            """
        )
        Test_PTableTest._connection.commit()

        ethalons = [
            [(0, 'id', 'INTEGER'), (1, 'a1', 'INTEGER'), (2, 'b1', 'REAL'), (3, 'c1', 'TEXT'), (4, 'd1', 'BLOB'), (5, 'e1', 'DATE'), (6, 'f1', 'TIMESTAMP')],
            [(0, 'a2', 'INTEGER'), (1, 'b2', 'INTEGER')],
            [(0, 'id', 'INTEGER'), (3, 'd3', 'BOOL')]
        ]
        results = []

        for name in ['test_wo_keys_1', 'test_wo_keys_2', 'test_wo_keys_3']:
            testTable = PTable(name, Test_PTableTest._connection)
            results.append(testTable.fetchColumnsWithoutForeignKeys())

        for pair in zip(ethalons, results):
            self.assertEqual(pair[0], pair[1], 'Failed with {}.'.format(pair))

if __name__ == '__main__':
    unittest.main()
